package com.lx.rabbitmq.four;

import com.lx.rabbitmq.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;

import java.util.UUID;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * 发布确认模式
 * 1、单个确认
 * 2、批量确认
 * 3、异步确认
 *
 * @author lixing
 * @date 2021/9/28 16:01
 */
public class ConfirmMessage {

    public static void main(String[] args) {

    }

    //单个确认
    public static void publishMessageIndividually() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //队列的声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            //开始时间
            long timeMillis = System.currentTimeMillis();
            //批量发消息
            for (int i = 0; i < 1000; i++) {
                String name = String.valueOf(i);
                channel.basicPublish("", queueName, null, name.getBytes());
                //单个消息确认
                boolean flag = channel.waitForConfirms();
                if (flag) {
                    System.out.println(i + "----发送成功");
                }
            }
            System.out.println((System.currentTimeMillis()) - timeMillis + "ms");//722ms

        }
    }

    //批量确认
    public static void publishMessageBatch() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //队列的声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            //开始时间
            long timeMillis = System.currentTimeMillis();
            //批量发消息
            for (int i = 0; i < 1000; i++) {
                String name = String.valueOf(i);
                channel.basicPublish("", queueName, null, name.getBytes());
                //判断达到100条批量确认一次
                if ((i + 1) % 100 == 0) {
                    //发布确认
                    channel.waitForConfirms();
                }
            }

            System.out.println((System.currentTimeMillis()) - timeMillis + "ms");//141ms
        }

    }

    //异步确认
    public static void publishMessageAsync() throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            //队列的声明
            String queueName = UUID.randomUUID().toString();
            channel.queueDeclare(queueName, true, false, false, null);
            //开启发布确认
            channel.confirmSelect();
            /**
             * 线程安全有序的一个哈希表，适用于高并发的情况
             * 1.轻松的将序号与消息进行关联
             * 2.轻松批量删除条目 只要给到序列号
             * 3.支持并发访问
             */
            ConcurrentSkipListMap<Long, String> outstandingConfirms = new ConcurrentSkipListMap<>();


            //准备消息监听器 监听消息的成功和失败
            /**
             * 确认收到消息的一个回调
             * 1.消息序列号
             * 2.true 可以确认小于等于当前序列号的消息
             * false 确认当前序列号消息
             */
            ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                if (multiple) {
                    //返回的是小于等于当前序列号的未确认消息 是一个 map
                    ConcurrentNavigableMap<Long, String> confirmed = outstandingConfirms.headMap(deliveryTag, true);
                    //清除该部分未确认消息
                    confirmed.clear();
                } else {
                    //只清除当前序列号的消息
                    outstandingConfirms.remove(deliveryTag);
                }

            };
            //消息确认失败 回调函数
            ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
                String message = outstandingConfirms.get(deliveryTag);
                System.out.println("发布的消息" + message + "未被确认，序列号" + deliveryTag);

            };
            /**
             * 添加一个异步确认的监听器
             * 1.确认收到消息的回调
             * 2.未收到消息的回调
             */
            channel.addConfirmListener(ackCallback, nackCallback);

            //开始时间
            long timeMillis = System.currentTimeMillis();
            //批量发消息
            for (int i = 0; i < 1000; i++) {
                String message = String.valueOf(i);
                /**
                 * channel.getNextPublishSeqNo()获取下一个消息的序列号
                 * 通过序列号与消息体进行一个关联
                 * 全部都是未确认的消息体
                 */
                outstandingConfirms.put(channel.getNextPublishSeqNo(), message);
                channel.basicPublish("", queueName, null, message.getBytes());
            }

            System.out.println((System.currentTimeMillis()) - timeMillis + "ms");//62ms
        }

    }

}
